package com.ls.rabbitmqService.topic;

import com.ls.rabbitmqService.utils.ConnectionFactoryUtil;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class ReceivedMessageTopic {


    private static final Logger logger = LoggerFactory.getLogger(ReceivedMessageTopic.class);

    public static void receiveMessage(String exchanges_name,String routingKey){
        Connection connection =null;
        Channel channel=null;
        try {
            connection=ConnectionFactoryUtil.getRabbitConnection();
            channel = connection.createChannel();
            channel.exchangeDeclare(exchanges_name, BuiltinExchangeType.TOPIC,true);
            //随机队列
            String queueName = channel.queueDeclare().getQueue();
            queueName="queue.topic.test";
            channel.queueBind(queueName,exchanges_name,routingKey);

            logger.info(String.format("ReceivedMessageTopic exchange:%s,queueName:%s,severity:%s",exchanges_name,queueName,routingKey));
            logger.info(" ReceivedMessageTopic wait.....");

            final Channel channelCopy=channel;
            Consumer consumer=new DefaultConsumer(channelCopy){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //采用手动确认
                    channelCopy.basicAck(envelope.getDeliveryTag(),false);
                    String message=new String(body,"utf-8");
                    if(logger.isDebugEnabled())
                        logger.debug(String.format("ReceivedMessageTopic [x4] routingKey:%s message:%s",envelope.getRoutingKey(),message));
                }
            };

            //这样会持续的接受消息  false 代表需要手动确认消息
            channel.basicConsume(queueName,false,consumer);

        } catch (IOException e) {
           logger.error(e.getMessage(),e);
        } finally{

        }

    }

    private static final String EXCHANGES_NAME="topic_logs";

    public static void main(String[] args){

        Connection connection =null;
        Channel channel=null;
        try {
            connection=ConnectionFactoryUtil.getRabbitConnection();
            channel = connection.createChannel();

            channel.exchangeDeclare(EXCHANGES_NAME, BuiltinExchangeType.TOPIC);
            String queueName = channel.queueDeclare().getQueue();
            String[] routingKes=new String[]{"*"};
            for(String routingKey:routingKes){
                channel.queueBind(queueName,EXCHANGES_NAME,routingKey);
                System.out.println(String.format("ReceiveLogsTopic4 exchange:%s,queueName:%s,severity:%s",EXCHANGES_NAME,queueName,routingKey));
            }

            System.out.println(" ReceiveLogsTopic4 wait.....");

            Consumer consumer=new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message=new String(body,"utf-8");
                    System.out.println(String.format("ReceivedMessageTopic [x4] routingKey:%s message:%s",envelope.getRoutingKey(),message));
                }
            };

            channel.basicConsume(queueName,true,consumer);

        } catch (IOException e) {
            e.printStackTrace();
        } finally{
            /*if(channel!=null){
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }*/

        }

    }
}

